/*
 * Copyright Debezium Authors.
 *
 * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
 */
package com.yifeng.repo.flink.data.transport.utils.queues;

import io.debezium.annotation.ThreadSafe;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.time.Duration;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.TimeUnit;

@ThreadSafe
public class ChangeEventQueue<T> {

    private static final Logger LOGGER = LoggerFactory.getLogger(ChangeEventQueue.class);

    private final int maxQueueSize;
    private final BlockingQueue<T> queue;

    private ChangeEventQueue(int maxQueueSize) {
        this.maxQueueSize = maxQueueSize;
        this.queue = new LinkedBlockingDeque<>(maxQueueSize);
    }

    public static class Builder<T> {

        private int maxQueueSize;

        public Builder<T> maxQueueSize(int maxQueueSize) {
            this.maxQueueSize = maxQueueSize;
            return this;
        }

        public ChangeEventQueue<T> build() {
            return new ChangeEventQueue<T>(maxQueueSize);
        }
    }

    /**
     * Enqueues a record so that it can be obtained via {@link #poll()}. This method
     * will block if the queue is full.
     *
     * @param record
     *            the record to be enqueued
     * @throws InterruptedException
     *             if this thread has been interrupted
     */
    public void enqueue(T record) throws InterruptedException {
        if (record == null) {
            return;
        }
        // The calling thread has been interrupted, let's abort
        if (Thread.interrupted()) {
            throw new InterruptedException();
        }
        doEnqueue(record);
    }


    protected void doEnqueue(T record) throws InterruptedException {
        if (LOGGER.isDebugEnabled()) {
            LOGGER.debug("Enqueuing source record '{}'", record);
        }
        // this will also raise an InterruptedException if the thread is interrupted while waiting for space in the queue
        queue.put(record);
    }

    /**
     * Returns the next batch of elements from this queue. May be empty in case no
     * elements have arrived in the maximum waiting time.
     *
     * @throws InterruptedException
     *             if this thread has been interrupted while waiting for more
     *             elements to arrive
     */
    public T take() throws InterruptedException {
        try {
            LOGGER.debug("taking records...");
            return queue.take();
        } finally {
        }
    }
    
    public T poll(long timeout, TimeUnit unit) throws InterruptedException {
        try {
            LOGGER.debug("polling records...");
            return queue.poll(timeout,unit);
        } finally {
        }
    }
    
    public T poll() throws InterruptedException {
        try {
            LOGGER.debug("polling records...");
            return queue.poll();
        } finally {
        }
    }

}
